package org.raymond.iworks.study.mq.rabbitmq.quorum;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

/**
 * 适用场景:
 * 仲裁队列是专门根据设计构建的。它们不是设计用来解决所有问题的。它们的预期用途是用于队列长期存在并且对系统操作的某些方面至关重要的拓扑结构，因此容错性和数据安全性比尽可能低的延迟和高级队列特性更重要。
 * 例如销售系统中的订单或选举系统中的投票，其中可能丢失的消息会对系统的正确性和功能产生重大影响。
 * 股票报价程序和即时消息传递系统从仲裁队列中受益较少或根本没有受益。
 * 发布者应该使用发布者确认，因为这是客户端与仲裁队列共识系统交互的方式。只有当已发布的消息成功复制到quorum节点并且在系统上下文中被认为是“安全的”时，才会发出发布者确认。
 * 消费者应该使用手动确认，以确保未成功处理的消息返回到队列，以便另一个消费者可以重新尝试处理。
 *
 * 不适用场景:
 * 队列的临时性质:瞬态或独占队列、高队列流失率(声明和删除率)
 * 尽可能低的延迟:由于其数据安全特性，底层共识算法具有固有的更高延迟
 * 当数据安全不是优先级时(例如，应用程序不使用手动确认和不使用发布者确认)
 * 非常长的队列积压(流可能更适合)
 *
 * 仲裁队列被设计为以延迟换取吞吐量
 *     由于仲裁队列在执行任何操作之前将所有数据保存到磁盘,因此建议使用最快的磁盘.
 * 仲裁队列还受益于消费者使用更高的预取值,以确保在确认通过系统时消费者不会挨饿,并允许及时传递消息.
 *     由于仲裁队列需要大量的磁盘I/O,它们的吞吐量会随着消息大小的增加而降低.
 *     与镜像队列一样,仲裁队列也受到集群大小的影响.仲裁队列的副本越多,它的吞吐量就越低,
 * 因为复制数据和达成一致需要做更多的工作
 *
 * 资源占用:
 *     仲裁队列针对数据安全和性能进行了优化,在稳定的工作负载下,它通常比经典镜像队列需要更多的资源(磁盘和RAM).
 * 每个仲裁队列进程维护队列中消息的内存索引,这对于每条消息至少需要32字节的元数据(如果返回消息或设置了TTL,则需要更多).
 * 因此,仲裁队列进程将为队列中的每30000条消息至少使用1MB(消息大小无关).您可以根据队列的数量(以及队列中的预期或最大消息数量)
 * 执行粗略计算.保持队列较短是保持低内存使用的最佳方法.如果队列由于某种原因变得很长,那么为所有队列设置最大队列长度是
 * 限制总内存使用的好方法.
 *     此外,给定节点上的仲裁队列共享所有操作的预写日志(WAL).WAL操作既存储在内存中,也写入磁盘.当当前的WAL文件达到预定义的限制时,
 * 它被刷新到磁盘上的WAL段文件中,系统将开始释放该批日志条目所使用的内存.随着时间的推移,随着消费者确认交付,片段文件被压缩.
 * 压缩是回收磁盘空间的过程
 *     缺省值为512MiB.这意味着在稳定负载期间,WAL表的内存占用可以达到512MiB.由于内存释放可能需要一些时间,我们建议RabbitMQ节点
 * 分配的内存至少是默认WAL文件大小限制的3倍.在高通量系统中需要更多.4倍是一个很好的起点.
 */
@Slf4j
public class NormalTask {
    public static void main(String[] args) throws Exception {
        String message = "to quorum..";
        try(Connection connection = NormalWorker.connection();
            Channel channel = connection.createChannel()){
            Map<String, Object> map = new HashMap<>();
            map.put("x-queue-type", "quorum");
            channel.queueDeclare(NormalWorker.QUEUE_NAME, true, false, false, map);
            channel.exchangeDeclare(NormalWorker.EXCHANGE_NAME, BuiltinExchangeType.TOPIC);

            for(int i=0; i<1; i++){
                channel.basicPublish(NormalWorker.EXCHANGE_NAME, NormalWorker.ROUTING_KEY, null, message.getBytes(StandardCharsets.UTF_8));
            }
            log.info("[x] sent: {}:{}", "", message);
        }
    }
}
